#include "RelOp.h"

void SelectFile::Run (DBFile &inFile, Pipe &outPipe, CNF &selOp, Record &literal) {

	Inputs *inp=  new Inputs(&inFile, &outPipe, &selOp, &literal);

	pthread_create(&thread, NULL, SelectFile_run, (void*)inp); 

}

void SelectFile::WaitUntilDone () {
	pthread_join (thread, NULL);
}

void* SelectFile::SelectFile_run(void* args)
{
	Record rec;
	Inputs *input = (Inputs*)args;    
	input->inFile->MoveFirst();    
	while(input->inFile->GetNext(rec, *(input->cnf), *(input->literal)))    
	{ 
		input->outPipe->Insert(&rec);   
	}  
	input->outPipe->ShutDown();     
	delete input; 
	input = NULL; 


}

void SelectFile::Use_n_Pages (int runlen) {

}



void SelectPipe::Run (Pipe &inPile, Pipe &outPipe, CNF &selOp, Record &literal) {

	Inputs *inp=  new Inputs(&inPile, &outPipe, &selOp, &literal);

	pthread_create(&thread, NULL, SelectPipe_run, (void*)inp);

}

void SelectPipe::WaitUntilDone () {
	pthread_join (thread, NULL);
}

void* SelectPipe::SelectPipe_run(void* args)
{
	Record rec;
	ComparisonEngine ceng;
	Inputs *input = (Inputs*)args;

	while(input->inPipe->Remove(&rec))
	{      
		if (ceng.Compare(&rec, (input->literal), (input->cnf)) )   
			input->outPipe->Insert(&rec); 

	}
	input->outPipe->ShutDown();
	delete input;
	input = NULL;


}

void Project::Run (Pipe &inPipe, Pipe &outPipe, int *keepMe, int numAttsInput, int numAttsOutput) 
{       
	Inputs *input = new Inputs(&inPipe, &outPipe, keepMe, numAttsInput, numAttsOutput);
	pthread_create(&thread, NULL, &Project_run, (void*) input); 
	return; 

}
void * Project::Project_run(void * args)
{        
	Inputs* input= (Inputs*)args;
	Record rec;              
	while(input->inPipe->Remove(&rec))
	{               
		rec.Project(input->AttsToKeep, input->numAttsToKeep, input->numAttsRec);
		input->outPipe->Insert(&rec);   
	}  
	input->outPipe->ShutDown();     
	delete input;     
	input = NULL; 
} 
void Project::WaitUntilDone() {    

	pthread_join(thread, 0); 
} 


void Sum::Run (Pipe &inPipe, Pipe &outPipe, Function &computeMe)
{      
	Inputs *input= new Inputs(&inPipe, &outPipe, &computeMe);
	pthread_create(&thread, NULL, &Sum_run,(void*)input);                 
	return; 
}  
void* Sum::Sum_run(void* args)
{  
	Inputs* input = (Inputs*)args;       
	Record rec;            
	double sum = 0;
	int i = 0;int a;       
	while(input->inPipe->Remove(&rec))        
	{
		i++;                 
		int ival = 0; 
		double dval = 0;                

		input->comp->Apply(rec, ival, dval);                
		sum += (ival + dval);               
	}

	//cast to char*   and create a file with single line of record and use sucknextrecord         
	Attribute att = {(char*)"sum", Double}; 
	FILE * sumFile = fopen("tmp_file", "w");        
	fprintf(sumFile, "%f|", sum);        
	fclose(sumFile);        
	sumFile = fopen("tmp_file", "r");        

	Schema sumSchema((char*)"tmp",1, &att);
	rec.SuckNextRecord(&sumSchema, sumFile);         
	fclose(sumFile);          

	input->outPipe->Insert(&rec);     

	input->outPipe->ShutDown();          
	delete input;     
	input = NULL; 
} 
void Sum::WaitUntilDone()
{ 

	pthread_join(thread, 0); 
} 


void WriteOut::Run (Pipe &inPipe, FILE *outFile, Schema &mySchema)
{
	Inputs *input= new Inputs(&inPipe, &mySchema, outFile);
	pthread_create(&thread, NULL, &WriteOut_run,(void*)input);
	return;
}
void * WriteOut::WriteOut_run(void * args) 
{  

	Inputs *input = (Inputs*)args;
	Record rec;      
	int count=0;          
	while(input->inPipe->Remove(&rec))        
	{                 

		count++;    
		int n = input->sch->GetNumAtts();     
		Attribute *atts = input->sch->GetAtts();    
		for (int i = 0; i < n; i++)    
		{               
			fprintf(input->file, "%s: ", atts[i].name);
			int pointer = ((int *) rec.bits)[i + 1];     
			fprintf(input->file, "[");    
			if (atts[i].myType == Int)  
			{                     
				int *myInt = (int *) &(rec.bits[pointer]);
				fprintf(input->file, "%d", *myInt);                         
			}                         
			else if (atts[i].myType == Double)                  
			{                    
				double *myDouble = (double *) &(rec.bits[pointer]);  
				fprintf(input->file, "%f", *myDouble);              
			}                  
			else if (atts[i].myType == String)     
			{                
				char *myString = (char *) &(rec.bits[pointer]);  
				fprintf(input->file, "%s", myString);                 }    
				fprintf(input->file, "]");             
				// print out a comma as needed to make things pretty               
				if (i != n - 1)                       
				{                                
					fprintf(input->file, ", ");         
				}        
		}             

		// end of for loop                
		fprintf(input->file, "\n");        
	}                   
	// no more records in the inPipe        
	cout<<"count is"<<count;         
	delete input;       
	input = NULL;
} 


void WriteOut::WaitUntilDone()
{

	pthread_join(thread, 0);
}

int DuplicateRemoval::runlen = 10; 
void DuplicateRemoval::Run(Pipe &inPipe, Pipe &outPipe, Schema &mySchema) {       
	if (runlen == -1)   
	{ 
		cerr << "\n set memeory pages";           
		exit(1);      
	}   
	Inputs *input = new Inputs(&inPipe,&outPipe,&mySchema);


	pthread_create(&thread, NULL, &DuplicateRemoval_run,  (void*)input);   
	return;
}  

void * DuplicateRemoval::DuplicateRemoval_run(void * args) 

{ 
	Inputs* input = (Inputs*)args;
	ComparisonEngine ceng;  
	Record prevRec, currRec;  
	OrderMaker sortOrder;   
	Pipe bigQPipe(100);    
	bool prevSeenRec = false;

	int n = input->sch->GetNumAtts();    
	Attribute *atts = input->sch->GetAtts();       
	sortOrder.numAtts = n;    
	for (int i = 0; i < n; i++)    
	{          
		sortOrder.whichAtts[i] = i;              
		sortOrder.whichTypes[i] = atts[i].myType;         
	}      

	BigQ bigQ(*(input->inPipe), bigQPipe, sortOrder, runlen); 


	while (bigQPipe.Remove(&currRec))      
	{           
		if (prevSeenRec == false)        
		{            
			prevRec.Copy(&currRec);  
			prevSeenRec= true;      
		}             
		if (ceng.Compare(&prevRec, &currRec, &sortOrder) != 0)       
		{     
			input->outPipe->Insert(&prevRec);
			prevRec.Copy(&currRec);    
		}         }          
		input->outPipe->Insert(&prevRec);      
		bigQPipe.ShutDown();     
		input->outPipe->ShutDown();          
		delete input;    
		input = NULL; 
} 
void DuplicateRemoval::Use_n_Pages(int n) {        
	runlen = n;
} 
void DuplicateRemoval::WaitUntilDone() {   
	pthread_join(thread, 0);
} 



int Join::runlen = 10;
void  Join::Run(Pipe &inPipeLe, Pipe &inPipeRi, Pipe &outPipeR, CNF &selOp, Record &literal, Schema& s1, Schema& s2)

{ 
	Inputs *input = new Inputs(&inPipeLe, &inPipeRi, &outPipeR, &selOp, &literal,&s1,&s2);
	pthread_create(&thread, NULL, &Join_run,  (void*)input);   
	return;
} 

void * Join::Join_run(void* args)
{
	Inputs* input = (Inputs*)args;
	OrderMaker orderL, orderR;  
	input->cnf->GetSortOrders(orderL, orderR);
	ComparisonEngine ceng;
	vector<Record*> recordR;
	Record *leftRec, *rightRec;
	leftRec = new Record();
	rightRec = new Record();

	if (orderL.numAtts == orderR.numAtts &&( ( orderL.numAtts > 0) && ( orderR.numAtts > 0))) 
	{        
		Pipe outL(100);
		Pipe  outR(100);
		Record r;
		BigQ *bigQR = new BigQ(*(input->inPipeR), outR, orderR, runlen); 
		while(outR.Remove(rightRec)) {
			recordR.push_back(rightRec);
			rightRec = new Record();
		}
		delete bigQR;
		int index = 0,tmpInd;
		bool looping = false;
		cout << "============\n";

		BigQ *bigQL = new BigQ(*(input->inPipeL), outL, orderL, runlen);   
		while(outL.Remove(leftRec)) {
			while(index < recordR.size() ){
				if(!looping){
					looping = true;
					tmpInd = index;
				}
				rightRec = recordR.at(index++);
				int diff = ceng.Compare(leftRec,&orderL,rightRec,&orderR);
				if(diff == 0) {
					int lefttotal = input->s1->GetNumAtts();
					int righttotal = input->s2->GetNumAtts();
					int numAttsToKeep = lefttotal + righttotal;
					int attsToKeep[numAttsToKeep];
					for(int k=0;k<lefttotal;k++)
					{
						attsToKeep[k]=k;
					}
					for(int k=lefttotal;k<numAttsToKeep;k++)
					{
						attsToKeep[k]=(k-lefttotal);
					}
					Record *tmp = new Record();
					tmp->MergeRecords(leftRec,rightRec,lefttotal,righttotal,attsToKeep,numAttsToKeep,lefttotal);
					input->outPipe->Insert(tmp);
				} else if(diff == 1){
					tmpInd++;
					continue;
				} else if(diff == -1) {
					index= tmpInd;
					looping = false;	
					break;
				} 
			}
		}

		delete bigQL;
	}
	//nested loop join
	else{
		while(input->inPipeR->Remove(rightRec)) {
			recordR.push_back(rightRec);
			rightRec = new Record();
		}
		while(input->inPipeL->Remove(leftRec)) {
			for(int index=0;index<recordR.size();index++) {
				rightRec = recordR.at(index);
				int diff = ceng.Compare(leftRec,&orderL,rightRec,&orderR);
				if(diff == 0){
					int lefttotal = input->s1->GetNumAtts();
					int righttotal = input->s2->GetNumAtts();
					int numAttsToKeep = lefttotal + righttotal;             
					int attsToKeep[numAttsToKeep];
					for(int k=0;k<lefttotal;k++)
					{
						attsToKeep[k]=k;
					}
					for(int k=lefttotal;k<numAttsToKeep;k++)
					{
						attsToKeep[k]=(k-lefttotal);
					}
					Record *tmp = new Record();
					tmp->MergeRecords(leftRec,rightRec,lefttotal,righttotal,attsToKeep,numAttsToKeep,lefttotal);
					input->outPipe->Insert(tmp);
				}
			}
		}

	}
	input->outPipe->ShutDown();
	delete input;    
	input = NULL;
}
void Join::Use_n_Pages(int n) {        
	runlen = n;
} 
void Join::WaitUntilDone() {   
	pthread_join(thread, 0);
}

int GroupBy::runlen = 10;
void GroupBy::Use_n_Pages(int n)
{
	//runlen = n;
}

void GroupBy::Run(Pipe& inPipe, Pipe& outPipe, OrderMaker& groupAtts, Function& computeMe)
{
	cout << "In function: " <<  &inPipe << "\n";	
	Inputs * input = new Inputs(&inPipe, &outPipe, &groupAtts, &computeMe);
	pthread_create(&thread, NULL, &GroupBy_run, (void*)input);
}

void GroupBy::WaitUntilDone()
{
	pthread_join(thread, 0);
}

void* GroupBy::GroupBy_run(void* args)
{
	Inputs* input = (Inputs*)args;
	Record *tmp = new Record();

	Pipe bigQOutPipe(100);
	//input->grpAttr->Print();

	BigQ *bigQ = new BigQ(*(input->inPipe), bigQOutPipe, *(input->grpAttr), runlen);
	Record rec;
	Record *currentRecord = new Record();
	ComparisonEngine ceng;
	double sum = 0.0;

	if(bigQOutPipe.Remove(&rec)) {
		currentRecord->Copy(&rec);
		int ival = 0; double dval = 0;
		input->func->Apply(rec, ival, dval);
		sum += (ival + dval);
	}

	while(bigQOutPipe.Remove(&rec))
	{
		if(ceng.Compare(currentRecord , &rec, input->grpAttr) == 0)
		{
			int ival = 0; double dval = 0;
			input->func->Apply(rec, ival, dval);
			sum += (ival + dval);

		}
		else
		{
			Attribute att = {(char*)"sumG", Double};
			Schema sumG_schema((char*)"tmp_grpby", 1, &att); 

			FILE * sumG_file = fopen("tmp_grpby_file","w");
			fprintf(sumG_file, "%f|", sum);
			fclose(sumG_file);
			sumG_file = fopen("tmp_grpby_file","r");
			Record sumSingleRec;
			sumSingleRec.SuckNextRecord(&sumG_schema, sumG_file);
			fclose(sumG_file);


			int totAttsToKeep = input->grpAttr->numAtts + 1;
			int attsToKeep[totAttsToKeep];
			attsToKeep[0] = 0; //for sumRec
			for(int i = 1; i < totAttsToKeep; i++)
			{
				attsToKeep[i] = input->grpAttr->whichAtts[i-1];
			}
			Record final_rec;
			final_rec.MergeRecords(&sumSingleRec, currentRecord , 1, totAttsToKeep - 1, attsToKeep, totAttsToKeep, 1);

			input->outPipe->Insert(&final_rec);

			//initialize sum from last unused record (if any)
			currentRecord->Copy(&rec);
			int ival = 0; double dval = 0;
			input->func->Apply(rec, ival, dval);
			sum = (ival + dval); 

			if(remove("tmp_grpby_file") != 0)
				perror("\nnot removed our temp file for groupby");
		}
	}
	Attribute att = {(char*)"sumG", Double};
	Schema sumG_schema((char*)"tmp_grpby", 1, &att);

	FILE * sumG_file = fopen("tmp_grpby_file","w");
	fprintf(sumG_file, "%f|", sum);
	fclose(sumG_file);
	sumG_file = fopen("tmp_grpby_file","r");
	Record sumSingleRec;
	sumSingleRec.SuckNextRecord(&sumG_schema, sumG_file);
	fclose(sumG_file);


	int totAttsToKeep = input->grpAttr->numAtts + 1;
	int attsToKeep[totAttsToKeep];
	attsToKeep[0] = 0; //for sumRec
	for(int i = 1; i < totAttsToKeep; i++)
	{
		attsToKeep[i] = input->grpAttr->whichAtts[i-1];
	}
	Record final_rec;
	final_rec.MergeRecords(&sumSingleRec, currentRecord , 1, totAttsToKeep - 1, attsToKeep, totAttsToKeep, 1);

	input->outPipe->Insert(&final_rec);

	if(remove("tmp_grpby_file") != 0)
		perror("\nnot removed our temp file for groupby");


	// Shut down the outpipe
	input->outPipe->ShutDown();
	delete input;
	delete bigQ;
	bigQ = NULL;
	input = NULL;
}




